package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(Constant.TOPIC_DB,Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,4,10014);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        // 1.读取kafak topic_db数据
        readOdsDb(tableEnv,Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
        // 2. 过滤出订单详情数据
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "   `data`['id'] id,\n" +
                "   `data`['order_id'] order_id,\n" +
                "   `data`['sku_id'] sku_id,\n" +
                "   `data`['sku_name'] sku_name,\n" +
                "   `data`['order_price'] order_price,\n" +
                "   `data`['sku_num'] sku_num,\n" +
                "   `data`['create_time'] create_time,\n" +
                "   `data`['split_total_amount'] split_total_amount,\n" +
                "   `data`['split_activity_amount'] split_activity_amount,\n" +
                "   `data`['split_coupon_amount'] split_coupon_amount ,\n" +
                "   ts\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table` = 'order_detail'\n" +
                "and `type` = 'insert'");

        tableEnv.createTemporaryView("order_detail",orderDetail);

        // 3.过滤订单信息表
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "   `data`['id'] id,\n" +
                "   `data`['user_id'] user_id,\n" +
                "   `data`['province_id'] province_id\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table` = 'order_info'\n" +
                "and `type` = 'insert'");

        tableEnv.createTemporaryView("order_info",orderInfo);
        // 4、 参与活动的订单
        Table orderDetailActivity = tableEnv.sqlQuery("select \n" +
                "   `data`['activity_rule_id'] activity_rule_id,\n" +
                "   `data`['activity_id'] activity_id,\n" +
                "   `data`['order_detail_id'] order_detail_id\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table` = 'order_detail_activity'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_detail_activity",orderDetailActivity);
        // 5、 参与优惠卷的订单
        Table orderDetailCoupon = tableEnv.sqlQuery("select \n" +
                "   `data`['coupon_id'] coupon_id,\n" +
                "   `data`['order_detail_id'] order_detail_id\n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table` = 'order_detail_coupon'\n" +
                "and `type` = 'insert'");

        tableEnv.createTemporaryView("order_detail_coupon",orderDetailCoupon);

        Table order = tableEnv.sqlQuery("select\n" +
                "  od.id,\n" +
                "  od.order_id,\n" +
                "  sku_id,\n" +
                "  oi.user_id,\n" +
                "  oi.province_id,\n" +
                "  activity_id,\n" +
                "  activity_rule_id,\n" +
                "  coupon_id,\n" +
                "  sku_name,\n" +
                "  order_price,\n" +
                "  sku_num,\n" +
                "  create_time,\n" +
                "  split_total_amount,\n" +
                "  split_activity_amount,\n" +
                "  split_coupon_amount,\n" +
                "  ts\n" +
                "from order_detail od \n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id\n" +
                "left join order_detail_activity oc1\n" +
                "on od.id = oc1.order_detail_id\n" +
                "left join order_detail_coupon oc2\n" +
                "on od.id = oc2.order_detail_id");

        // 创建Kafka对应的表
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_ORDER_DETAIL+"(\n" +
                "  id String,\n" +
                "  order_id  String,\n" +
                "  sku_id  String,\n" +
                "  user_id String,\n" +
                "  province_id String,\n" +
                "  activity_id String,\n" +
                "  activity_rule_id  String,\n" +
                "  coupon_id String,\n" +
                "  sku_name  String,\n" +
                "  order_price String,\n" +
                "  sku_num String,\n" +
                "  create_time String,\n" +
                "  split_total_amount  String,\n" +
                "  split_activity_amount String,\n" +
                "  split_coupon_amount String,\n" +
                "  ts  Bigint ,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED \n" +
                ")"+ SQLUtil.getUpsertKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));

        // 写入
        /**
         * 如果sql用了left join right join full join 就回产生回撤流，如果要写入kafka必须要用
         * upsert kafka链接器，如果用了upsert kafka，必须要有主键 必须要有主键 必须要有主键
         */
        order.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL).execute();

    }

}
